package com.lcq.demo.listener;

import com.lcq.demo.common.enums.QueueEnum;
import com.lcq.demo.common.utils.RedisUtil;
import com.lcq.demo.service.EmailService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import software.amazon.awssdk.core.exception.AbortedException;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sqs.SqsClient;
import software.amazon.awssdk.services.sqs.model.DeleteMessageRequest;
import software.amazon.awssdk.services.sqs.model.Message;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 邮箱验证码队列监听器
 * @author 240701005
 */
@Component
@Slf4j
public class VerifyCodeListener {

    private static final String SUBJECT = "E-Mail 验证码";

    private ExecutorService executorService;

    @Resource
    private RedisUtil redisUtil;

    @Resource
    private EmailService emailService;

    @PostConstruct
    private void init() {
        executorService = Executors.newSingleThreadExecutor();
        executorService.submit(this::listenForMessages);
    }

    @PreDestroy
    public void close() {
        log.info("shutdown 1");
        executorService.shutdown();
    }

    private void listenForMessages() {
        SqsClient sqsClient = SqsClient.builder().region(Region.US_EAST_1).build();
        while (!Thread.currentThread().isInterrupted()) {
            try {
                ReceiveMessageRequest receiveMessageRequest = ReceiveMessageRequest.builder()
                        .queueUrl(QueueEnum.TEST_QUEUE_URL.url)
                        .maxNumberOfMessages(10)
                        .waitTimeSeconds(20)
                        .build();
                ReceiveMessageResponse response = sqsClient.receiveMessage(receiveMessageRequest);
                List<Message> messages = response.messages();
                if (!messages.isEmpty()) {
                    for (Message message : messages) {
                        try {
                            processMessage(message);
                            sqsClient.deleteMessage(DeleteMessageRequest.builder()
                                    .queueUrl(QueueEnum.TEST_QUEUE_URL.url)
                                    .receiptHandle(message.receiptHandle())
                                    .build());
                        } catch (Exception e) {
                            log.error("处理消息出错--！: {}", message.messageId(), e);
                        }
                    }
                }
            } catch (AbortedException e) {
                log.error("接收消息操作被中断--！", e);
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("接收消息失败--！", e);
            }
        }
    }

    private void processMessage(Message msg) {
        try {
            String msgString = msg.body();
            String email = msgString.split(":")[0];
            String verifyCode = msgString.split(":")[1];
            redisUtil.setNx(redisUtil.buildKey(email, "verifyCode"), verifyCode, 5L, TimeUnit.MINUTES);
            emailService.sendEmail(email, SUBJECT, verifyCode);
        } catch (Exception e) {
            log.error("Error processing message: {}", msg.messageId(), e);
        }
    }
}
